package com.googlecode.xcarve.netty.handler;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

import java.util.Map;
import java.util.concurrent.ExecutorService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Maps;
import com.googlecode.xcarve.enums.MessageType;
import com.googlecode.xcarve.netty.server.ServiceExport;
import com.googlecode.xcarve.netty.server.ServiceProxy;
import com.googlecode.xcarve.object.Demo;
import com.googlecode.xcarve.object.Header;
import com.googlecode.xcarve.object.NettyMessage;
import com.googlecode.xcarve.object.RPCReponse;
import com.googlecode.xcarve.object.RPCRequest;

/**
 * @author xinyun.d
 * 2014年8月28日
 */
public class NettyServerHandler extends ChannelHandlerAdapter{
	static final Logger logger = LoggerFactory.getLogger(NettyServerHandler.class); 
	static String[] whiteList = {"127.0.0.1"};
	public static final Map<String,Object> map = Maps.newHashMap();
	private final ExecutorService threadPool ;
	private final ServiceExport serviceContext ;
	
 
	
	public NettyServerHandler(ExecutorService threadPool){
		this.threadPool = threadPool;
		serviceContext = ServiceExport.getInstance();
	}
	
	@Override
	public void channelRead(final ChannelHandlerContext ctx, Object msg)
			throws Exception {
		final Channel channel = ctx.channel();
		final NettyMessage nettyMsg = (NettyMessage)msg;
		if(nettyMsg.getHeader()==null){
			throw new RuntimeException("invalid request message.");
		}
		final Object obj = nettyMsg.getBody();
		final RPCRequest request = (RPCRequest)obj;
		if(request.getEvent() == RPCRequest.HEART_BEAT_ENVENT){
			RPCReponse response = new RPCReponse(request.getSeqId());
			response.setStatus(true);
			response.setResult("hello client. now timestamp is "+System.currentTimeMillis());
			nettyMsg.setBody(response);
			channel.writeAndFlush(nettyMsg);
			return;
		}
		threadPool.submit(new Runnable() {
			
			@Override
			public void run() {
				if(obj instanceof RPCRequest){
					RPCReponse response = new RPCReponse(request.getSeqId());
				    Object result = null;
				    try {
				    	ServiceProxy proxy = serviceContext.getServiceProxy(request);
				    	if(proxy == null){
				    		response.setStatus(false);
				    		response.setMessage("service not found on serverSide！");
				    	}else{
//				    		logger.info("from clinet request={}",request);
				    		result = proxy.invoke(request);
				    	}
					} catch (Throwable e) {
						logger.error("invoke service error.",e);
					}
				    response.setResult(result);
					nettyMsg.setBody(response);
					channel.writeAndFlush(nettyMsg);
				}else{
					channel.writeAndFlush(buildHandshakeResult());
				}
			}
		});

 
	}
	
	
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception {
//		nodeCheck.remove(ctx.channel().remoteAddress().toString());
		ctx.close();
//		logger.error("catched exception. ",cause);
	}
	
	static NettyMessage buildHandshakeResult(){
		NettyMessage msg = new NettyMessage();
		Header header = new Header();
		header.setCodecType(MessageType.HANDSHAK_RESP.value);
		msg.setHeader(header);
		msg.setBody(new Demo());
		return msg;
	}
	
}
